// Copyright (c) 2020-present, INSPUR Co, Ltd. All rights reserved.
// This source code is licensed under Apache 2.0 License.

#include <ctime>
#include <memory>
#include <string>
#include <thread>

#include "db/db_test_util.h"
#include "db/memtable.h"
#include "db/range_del_aggregator.h"
#include "port/stack_trace.h"
#include "rangearena/range_arena.h"
#include "rocksdb/memtablerep.h"
#include "rocksdb/slice_transform.h"

namespace rocksdb {

class TestComp : public MemTableRep::KeyComparator {
 public:
  ~TestComp() {}

  DecodedType decode_key(const char* b) const{
    Slice buf = GetLengthPrefixedSlice(b);
    Slice mvccKey = Slice(buf.data(), buf.size() - 8);
    const char ts_size = mvccKey[mvccKey.size() - 1];
    if ((size_t)ts_size >= mvccKey.size()) {
      return 0;
    }
    return mvccKey;
  }

  int operator()(const char* a, const char* b) const {
    Slice sa = GetLengthPrefixedSlice(a);
    Slice sb = GetLengthPrefixedSlice(b);
    return sa.compare(sb);
  }

  int operator()(const char* prefix_len_key, const Slice& key) const {
    Slice buf = GetLengthPrefixedSlice(prefix_len_key);
    Slice mvccKey = Slice(buf.data(), buf.size() - 9);
    return mvccKey.compare(key);
  }
};

class InlineartTest : public DBTestBase {
 public:
  InlineartTest() : DBTestBase("/InlineartTest") {}
};

const char* genMVCCKey1(std::string key) {
  uint32_t keySize = key.size() + 1 + 8;

  const uint32_t encoded_len = VarintLength(keySize) + keySize + 1 + 8 + VarintLength(0);
  char* ret = new char[encoded_len];
  memset(ret, 0, encoded_len);

  char* p = EncodeVarint32(ret, keySize);
  memcpy(p, key.data(), key.size());
  p += key.size() + 1;
  uint64_t packed = PackSequenceAndType(10000, kTypeValue);
  EncodeFixed64(p, packed);
  p += 8;
  p = EncodeVarint32(p, 0);

  return ret;
}

void insertAndGet1(int tail, InlineART<const MemTableRep::KeyComparator &>* list) {
  bool res;
  std::cout << "Enter the loop" << std::endl;
  // Test the duplicate keys under stress
  SequenceNumber seq = 10000000;
  int record_sum = 500000;
  for (int i = 0; i < record_sum; i++) {
    seq++;
    
    std::string a = std::to_string(seq) + std::to_string(tail);
    const char* buf =  genMVCCKey1(a);

    VersionNode* cur = new VersionNode();
    cur->CASSetKey(nullptr, (void*)buf);
    res = list->Insert((const char*)cur);
    ASSERT_TRUE(res);
  }

  std::cout << "-----------insert all this thread records ok." << std::endl;
}

TEST_F(InlineartTest, ConcInsert) {

InternalKeyComparator cmp(BytewiseComparator());
MemTable::KeyComparator comparator_(cmp);
const MemTableRep::KeyComparator &cmp_= comparator_;
  InlineART<const MemTableRep::KeyComparator &>* list= new InlineART<const MemTableRep::KeyComparator &>(cmp_);

  std::thread test1(insertAndGet1, 0, list);
  std::thread test2(insertAndGet1, 1, list);
  std::thread test3(insertAndGet1, 2, list);
  std::thread test4(insertAndGet1, 3, list);
  std::thread test5(insertAndGet1, 4, list);
  std::thread test6(insertAndGet1, 5, list);
  std::thread test7(insertAndGet1, 6, list);
  std::thread test8(insertAndGet1, 7, list);

  test1.join();
  test2.join();
  test3.join();
  test4.join();
  test5.join();
  test6.join();
  test7.join();
  test8.join();
  std::cout << "+++++++++ all records insert ok.!!!!" << std::endl;

  InlineART<const MemTableRep::KeyComparator &>::Iterator* iter = list->GetIterator();
  iter->SeekToFirst();
  size_t totalKVNum = 0;
  while (iter->Valid()) {
    totalKVNum++;
    iter->Next();
  }
  std::cout << "-----------insert all records[" << totalKVNum << "] ok."
            << std::endl;
}

}  // namespace rocksdb

int main(int argc, char** argv) {
  rocksdb::port::InstallStackTraceHandler();
  ::testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
}